package com.casic.mqtt;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.PostConstruct;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import com.casic.exception.AopException;
import com.casic.kafka.service.KafkaService;

import lombok.extern.slf4j.Slf4j;

/**
 * mqtt 消费者类
 * @author 朵朵
 *
 */
@Slf4j
@Configuration
public class MqttConsumer implements MqttCallbackExtended{

	ExecutorService pool = Executors.newFixedThreadPool(1000);
	
	MqttClient client;
	
	MqttConnectOptions options;

	@Value("${mqtt.broker.serverUri}")
	private String MqttUrl;

	@Value("${mqtt.topic}")
	protected String topic;

	@Value("${mqtt.qos}")
	protected int qos;

	@Value("${mqtt.broker.cleansession}")
	private boolean cleansession;

	@Value("${mqtt.broker.automaticreconnect}")
	private boolean automaticreconnect;
	
	/**
	 * 初始化mqtt消费者
	 */
	@PostConstruct
	public void init() {
		try {
			client = new MqttClient(MqttUrl, java.util.UUID.randomUUID().toString(), new MemoryPersistence());
			log.info("初始化mqtt客户端"+client);
			
			log.info("配置mqtt客户端配置信息");
	    	options = new MqttConnectOptions();
			options.setCleanSession(cleansession);
			options.setConnectionTimeout(30);
			options.setKeepAliveInterval(15);
			options.setAutomaticReconnect(automaticreconnect);
			options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
			
			client.setCallback(this);
			client.connect(options);
			client.subscribe(topic, qos);
			log.info("初始化mqtt客户端！");
		} catch (MqttSecurityException e) {
			log.info("初始化mqtt客户端异常：MqttSecurityException！\n"+e.getLocalizedMessage());
			throw new AopException(e.getLocalizedMessage());
		} catch (MqttException e) {
			log.info("初始化mqtt客户端异常：MqttException！\n"+e.getLocalizedMessage());
			throw new AopException(e.getLocalizedMessage());
		}

	}
	
	
	 /**
	  * 注入Kafka的Service
	  */
	 @Autowired
	 private KafkaService kafkaService;

	/**
	 * 断开链接，重连操作
	 */
	public void connectionLost(Throwable cause) {
		log.info("[MQTT] 连接断开，1S之后尝试重连...");
		while (true) {
			try {
				if (!client.isConnected()) {
					log.info("重连mqtt客户端");
					client.setCallback(this);
					client.connect(options);
				}
				if (client.isConnected()) {
					log.info("mqtt客户端已重新链接");
					break;
				}
			} catch (MqttException e) {
				throw new AopException(e.getLocalizedMessage());
			}

		}
	}
	

	/**
	 * 连接成功
	 */
	@Override
	public void connectComplete(boolean reconnect, String serverURI) {
		log.info("开始订阅.............");
		try {
			client.subscribe(topic, qos);
			log.info("订阅主题："+topic+",qos:"+qos);
		} catch (MqttException e) {
			log.info("订阅主题："+topic+",qos:"+qos+"异常："+e.getLocalizedMessage());
			throw new AopException(e.getLocalizedMessage());
		}
		log.info("已经订阅.........");
	}

	/**
	 * 收到新的消息推送
	 */
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		log.info("Consumer接收到的数据\n topic:" + topic + "\n message:" + message.toString());
		
		String string = message.toString();
		log.info("检查消息格式");
		if(string.startsWith("{")&&string.endsWith("}")) {
			log.info("----------------------数据格式正确---------------------");
			kafkaService.writeToKafka(string);
		}else{
			log.info("----------------------数据格式不正确---------------------");
		}

	}

	/**
	 * 消息处理结束
	 */
	public void deliveryComplete(IMqttDeliveryToken token) {
		log.info("消息处理结束时调用");
	}
}
